/*
 * Copyright (c) 2006-2007 Terracotta, Inc. All rights reserved.
 */
package com.google.zahra.workmanager;

// Terracotta cluster events packages
import com.google.zahra.Master;

import com.tc.cluster.DsoCluster;
import com.tc.cluster.DsoClusterEvent;
import com.tc.cluster.DsoClusterListener;
import com.tc.injection.annotations.InjectedDsoInstance;

import org.terracotta.workmanager.pipemanager.Pipe;
import org.terracotta.workmanager.pipemanager.PipeManager;
import org.terracotta.workmanager.routing.RoutableWorkItem;
import org.terracotta.workmanager.worker.Worker;
import org.terracotta.workmanager.worker.WorkerTask;

import java.util.concurrent.TimeUnit;

import commonj.work.WorkException;

/**
 * Worker that has an is aware of routing. Each instance of the
 * RoutingAwareWorker class gets a routing ID from the Terracotta server and
 * gets a unique work queue mapped to this routing ID.
 *
 * @author Terracotta, Inc. modified by arthurkok@google.com ((Arthur) Kai Hei
 *         Kok)
 */
public class ZahraWorker implements Worker, DsoClusterListener {

  // Regardless of how many injections are added,
  // only one DsoCluster instance is instantiated per node.
  @InjectedDsoInstance
  private DsoCluster cluster;

  // The pipe for this worker
  private Pipe<RoutableWorkItem<String>, String> pipe;

  private PipeManager<String> pipeManager;

  private volatile boolean isRunning = true;

  private String routingID;

  private int myOs;

  private CommandChecker commandThread;
  
  private WorkerTask<RoutableWorkItem<String>> task;

  public static final int WORKER_TIMEOUT_IN_SECONDS = 5;

  /**
   * Constructor creates a new {@code DynamicWorker} and adds a listener to the
   * Terracotta cluster
   *
   * @param myOs The OS of this worker
   */
  public ZahraWorker(final int myOs) {
    this.myOs = myOs; // 0 = UNSPEC, 1 = WIN, 2 = MAC 4 = LINUX;
    cluster.addClusterListener(this);
    commandThread = new CommandChecker();
    commandThread.start();
  }

  /*
   * (non-Javadoc)
   *
   * @see
   * com.tc.cluster.DsoClusterListener#nodeJoined(com.tc.cluster.DsoClusterEvent
   * )
   */
  @Override
  public void nodeJoined(DsoClusterEvent arg0) {
    System.out.println("THE NODE " + arg0.getNode().getId() + " HAS JOINED");
  }

  /*
   * (non-Javadoc)
   *
   * @see
   * com.tc.cluster.DsoClusterListener#nodeLeft(com.tc.cluster.DsoClusterEvent)
   */
  @Override
  public void nodeLeft(DsoClusterEvent arg0) {
    System.out.println("THE NODE " + arg0.getNode().getId() + " HAS LEFT");
  }

  /*
   * (non-Javadoc)
   *
   * @seecom.tc.cluster.DsoClusterListener#operationsDisabled(com.tc.cluster.
   * DsoClusterEvent)
   */
  @Override
  public void operationsDisabled(DsoClusterEvent arg0) {
    System.out.println("OPERATIONS DISABLED");
    stop();
  }

  /*
   * (non-Javadoc)
   *
   * @seecom.tc.cluster.DsoClusterListener#operationsEnabled(com.tc.cluster.
   * DsoClusterEvent)
   */
  @Override
  public void operationsEnabled(DsoClusterEvent arg0) {
    System.out.println("OPERATIONS ENALBED");

    // Checks if this node has already been previously registered
    if (pipeManager == null) {
      thisNodeId(arg0.getNode().getId());
    }

  }

  /**
   * Starts the worker.
   *
   *  It first waits for registration from the pipe manager. Then it peeks the
   * pending work queue and retrieves the {@code WorkItem}. It creates a {@code
   * WorkerTask} and runs it. It is only after the {@code WorkerTask} completes
   * successfully that the corresponding {@code WorkItem} is removed from the
   * queue. This ensures that the pending tasks is there to recover if the
   * worker died in the middle of running a task.
   *
   */
  public void start() throws WorkException {   
    
    while (true) {
      if (isRunning) {

        if (pipeManager == null) {
          // Waits until the pipe manager had registered this worker
          waitForRegistration();          
          pipeManager.get(routingID).setStatus("RUNNING");
        }       
        
        final RoutableWorkItem<String> workItem;
        try {

          // Peeks the queue without removing the item from the queue
          //workItem = pipe.getPendingWorkQueue().peek();
          workItem = pipe.getPendingWorkQueue().polledPeek(WORKER_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);

          // Shuts down the worker if it waited too long
          /*
           * workItem =
           * pipe.getPendingWorkQueue().poll(WORKER_TIMEOUT_IN_SECONDS,
           * TimeUnit.SECONDS); if (workItem == null) {
           * System.out.println("Worker has waited " +
           * String.valueOf(WORKER_TIMEOUT_IN_SECONDS) +
           * " seconds for new work. Stopping now."); isRunning = false; }
           */

        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new WorkException(e);
        }

        if (workItem != null) {
          
          pipeManager.get(routingID).setStatus("RUNNING");

          // Creates and runs a new WorkerTask with the WorkItem
          
          task = new WorkerTask<RoutableWorkItem<String>>(workItem, pipe.getCompletedWorkQueue());
          task.run();
          task = null;

          // After the task finishes running, the task is them removed from the
          // queue
          try {
            pipe.getPendingWorkQueue().take();
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new WorkException(e);
          }

          System.out.println(
              String.format("Pending: %d, Completed: %d", pipe.getPendingWorkQueue().size(),
                  pipe.getCompletedWorkQueue().size()));
        }
        else {
          pipeManager.get(routingID).setStatus("IDLE");
        }
      }
    }
  }

  /**
   * Stops the worker
   */
  public void stop() {
    isRunning = false;
    if (task != null) {
      task.stop();
    }
    //pipeManager.removePipeFor(routingID);
    commandThread.requestStop();
    System.out.println("Bye.");
    System.exit(0);
  }

  /**
   * Sets the id of this client and registers it with pipe manager
   *
   * @param nodeId The node id of this client
   */
  public synchronized void thisNodeId(final Object nodeId) {
    System.out.println("worker node id: " + nodeId);
    routingID = (String) nodeId;

    pipeManager =
        new PipeManager<String>(Master.PIPE_FACTORY, routingID, PipeManager.Initiator.WORKER, myOs);

    pipe = pipeManager.getOrCreatePipeFor(routingID);
    notifyAll(); // worker is registered - notify the worker
  }

  /**
   * Blocks until it receives a notification from the pipe manager that client
   * is registered and a pipe has been created
   */
  private synchronized void waitForRegistration() {
    try {
      if (pipe == null) {
        System.out.println("waiting for registration");
        wait();
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private void runCommand(String command) {
    if (command.equals("STOP")) {
      System.out.println("Stopping worker...");
      pipeManager.get(routingID).setStatus("STOPPED");
      isRunning = false;
    } else if (command.equals("START")) {
      System.out.println("Starting worker...");
      pipeManager.get(routingID).setStatus("RUNNING");
      isRunning = true;
    } else if (command.equals("SHUTDOWN")) {
      System.out.println("Shutting down worker...");
      pipeManager.get(routingID).setStatus("SHUTDOWNED");
      stop();
    }
  }

  class CommandChecker extends Thread {

    // Must be volatile:
    private volatile boolean stop = false;
    String command;

    @Override
    public void run() {
      while (!stop) {

        if (pipe == null) {
          continue;
        }
        
        try {
          command = pipe.getCommandQueue().take();
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }

        if (command != null) {
          System.out.println("Received command: " + command);
          runCommand(command);
        }
      }

      if (stop) {
        System.out.println("Detected stop");
      }
    }

    public void requestStop() {
      stop = true;
    }

  }

}
